
package etcdserver

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"sort"
	"strconv"
	"strings"
	"time"

	"go.etcd.io/etcd/api/v3/version"
	"go.etcd.io/etcd/client/pkg/v3/types"
	"go.etcd.io/etcd/server/v3/etcdserver/api/membership"

	"github.com/coreos/go-semver/semver"
	"go.uber.org/zap"
)

// isMemberBootstrapped尝试检查给定集群中的给定成员是否已被引导
//  isMemberBootstrapped（） 函数，该函数会请求远端节点，然后检测当前集群中是否己存在相同的节点
func isMemberBootstrapped(lg *zap.Logger, cl *membership.RaftCluster, member string, rt http.RoundTripper, timeout time.Duration) bool {
	// 从远端节点获取集群信息，并返回 RaftCluster 实例
	rcl, err := getClusterFromRemotePeers(lg, getRemotePeerURLs(cl, member), timeout, false, rt)
	if err != nil {
		return false
	}
	// 当前节点的 id
	id := cl.MemberByName(member).ID
	// 根据 id 获取对应的 Member 实例
	m := rcl.Member(id)
	if m == nil {
		return false
	}
	if len(m.ClientURLs) > 0 {	// 检测己启动的相同节点是否对 Client 暴露了 URL
		return true
	}
	return false
}

// GetClusterFromRemotePeers获取一组表示etcd对等点的URL，而
// 尝试通过访问其中一个URL上的成员终结点来构建集群。使用第一个提供响应的URL。如果没有URL提供
// 响应，或者无法从收到的
// 响应成功创建群集，则返回错误。
// 每个请求都有10秒的超时时间。由于TTL的上限为5s，
// 10秒就足以满足建筑连接和完工要求。
func GetClusterFromRemotePeers(lg *zap.Logger, urls []string, rt http.RoundTripper) (*membership.RaftCluster, error) {
	return getClusterFromRemotePeers(lg, urls, 10*time.Second, true, rt)
}

// GetClusterFromRemotePeers （） 函数底层通过调用 getClusterFromRemotePeers（） 函数实现， 其
// 中会从集群中其他节点请求当前集群的信息，然后封装成 RaftCluster 实例返回
func getClusterFromRemotePeers(lg *zap.Logger, urls []string, timeout time.Duration, logerr bool, rt http.RoundTripper) (*membership.RaftCluster, error) {
	if lg == nil {
		lg = zap.NewNop()
	}
	cc := &http.Client{	// 创建一个 HTTP 客户端
		Transport: rt,
		Timeout:   timeout,
	}
	for _, u := range urls {	// 请求集群中其他节点，获取集群信息，如果发生错误，则换一个节点请求
		// 请求集群中其他节点提供的 HTTP 接口（ members ）
		addr := u + "/members"
		resp, err := cc.Get(addr)
		if err != nil {
			if logerr {
				lg.Warn("failed to get cluster response", zap.String("address", addr), zap.Error(err))
			}
			continue
		}
		// 读取响应，响应内容是 JSON 格式，
		b, err := io.ReadAll(resp.Body)
		resp.Body.Close()
		if err != nil {
			if logerr {
				lg.Warn("failed to read body of cluster response", zap.String("address", addr), zap.Error(err))
			}
			continue
		}
		var membs []*membership.Member
		// 解析 JSON 得到 Member 实例数组，
		if err = json.Unmarshal(b, &membs); err != nil {
			if logerr {
				lg.Warn("failed to unmarshal cluster response", zap.String("address", addr), zap.Error(err))
			}
			continue
		}
		// 从响应头中获取集群 ID ，
		id, err := types.IDFromString(resp.Header.Get("X-Etcd-Cluster-ID"))
		if err != nil {
			if logerr {
				lg.Warn(
					"failed to parse cluster ID",
					zap.String("address", addr),
					zap.String("header", resp.Header.Get("X-Etcd-Cluster-ID")),
					zap.Error(err),
				)
			}
			continue
		}

		// 检查成员的长度
		// 如果成员存在，则准备并返回raft群集
		// 如果成员不存在，则形成的raft群集将是
		// 无效的空群集，因此返回无法获取raft群集成员(s)从给定URL错误
		if len(membs) > 0 {	// 创建 RaftCluster 实例并返回
			return membership.NewClusterFromMembers(lg, id, membs), nil
		}
		return nil, fmt.Errorf("failed to get raft cluster member(s) from the given URLs")
	}
	return nil, fmt.Errorf("could not retrieve cluster information from the given URLs")
}

// getRemotePeerURLs返回群集中远程成员的对等URL。返回的列表按升序字典顺序排序。
// getRemotePeerURLs（） 函数会过滤掉当前节点提供的 URL 地址， 然后排序集群中其他节点
//暴露的 URL 地址并返回
func getRemotePeerURLs(cl *membership.RaftCluster, local string) []string {
	us := make([]string, 0)
	for _, m := range cl.Members() {
		if m.Name == local {	// 过滤：本地节点
			continue
		}
		us = append(us, m.PeerURLs...)
	}
	sort.Strings(us)	// 对其他节点的 URL 地址进行排序
	return us
}

// getMembersVersions返回给定群集中成员的版本。
// 返回的映射键为成员ID。返回的映射值
// 是Sever versions字符串，包括服务器和群集。
// 如果无法获取成员的版本，则密钥为零。
func getMembersVersions(lg *zap.Logger, cl *membership.RaftCluster, local types.ID, rt http.RoundTripper) map[string]*version.Versions {
	members := cl.Members()
	vers := make(map[string]*version.Versions)
	for _, m := range members {
		if m.ID == local {
			cv := "not_decided"
			if cl.Version() != nil {
				cv = cl.Version().String()
			}
			vers[m.ID.String()] = &version.Versions{Server: version.Version, Cluster: cv}
			continue
		}
		ver, err := getVersion(lg, m, rt)
		if err != nil {
			lg.Warn("failed to get version", zap.String("remote-member-id", m.ID.String()), zap.Error(err))
			vers[m.ID.String()] = nil
		} else {
			vers[m.ID.String()] = ver
		}
	}
	return vers
}

// allowedVersionRange决定本地服务器可以加入的群集的可用版本范围；
// 如果降级启用状态为true，则版本窗口为[oneMinorHigher，oneMinorHigher]
// 如果未启用降级，则版本窗口为[MinClusterVersion，localVersion]
func allowedVersionRange(downgradeEnabled bool) (minV *semver.Version, maxV *semver.Version) {
	minV = semver.Must(semver.NewVersion(version.MinClusterVersion))
	maxV = semver.Must(semver.NewVersion(version.Version))
	maxV = &semver.Version{Major: maxV.Major, Minor: maxV.Minor}

	if downgradeEnabled {
		// Todo:处理从更高主版本降级的情况（例如从v4.0降级到v3.x）
		maxV.Minor = maxV.Minor + 1
		minV = &semver.Version{Major: maxV.Major, Minor: maxV.Minor}
	}
	return minV, maxV
}

// isCompatibleWithCluster如果本地成员的版本与当前运行的群集的
// 兼容，则返回true。
// 当群集中至少一个其他成员的
// 群集版本在[MinV，MaxV]范围内时，该版本视为兼容没有已知成员的群集版本
// 超出范围。
// 我们设置了此规则，因为当本地成员加入时，另一个成员可能脱机。
func isCompatibleWithCluster(lg *zap.Logger, cl *membership.RaftCluster, local types.ID, rt http.RoundTripper) bool {
	vers := getMembersVersions(lg, cl, local, rt)
	minV, maxV := allowedVersionRange(getDowngradeEnabledFromRemotePeers(lg, cl, local, rt))
	return isCompatibleWithVers(lg, vers, local, minV, maxV)
}

func isCompatibleWithVers(lg *zap.Logger, vers map[string]*version.Versions, local types.ID, minV, maxV *semver.Version) bool {
	var ok bool
	for id, v := range vers {
		// 忽略与本地版本的比较
		if id == local.String() {
			continue
		}
		if v == nil {
			continue
		}
		clusterv, err := semver.NewVersion(v.Cluster)
		if err != nil {
			lg.Warn(
				"failed to parse cluster version of remote member",
				zap.String("remote-member-id", id),
				zap.String("remote-member-cluster-version", v.Cluster),
				zap.Error(err),
			)
			continue
		}
		if clusterv.LessThan(*minV) {
			lg.Warn(
				"cluster version of remote member is not compatible; too low",
				zap.String("remote-member-id", id),
				zap.String("remote-member-cluster-version", clusterv.String()),
				zap.String("minimum-cluster-version-supported", minV.String()),
			)
			return false
		}
		if maxV.LessThan(*clusterv) {
			lg.Warn(
				"cluster version of remote member is not compatible; too high",
				zap.String("remote-member-id", id),
				zap.String("remote-member-cluster-version", clusterv.String()),
				zap.String("minimum-cluster-version-supported", minV.String()),
			)
			return false
		}
		ok = true
	}
	return ok
}

// getVersion通过其
// pee返回给定成员的版本如果无法获取版本，则返回最后一个错误。
func getVersion(lg *zap.Logger, m *membership.Member, rt http.RoundTripper) (*version.Versions, error) {
	cc := &http.Client{
		Transport: rt,
	}
	var (
		err  error
		resp *http.Response
	)

	for _, u := range m.PeerURLs {
		addr := u + "/version"
		resp, err = cc.Get(addr)
		if err != nil {
			lg.Warn(
				"failed to reach the peer URL",
				zap.String("address", addr),
				zap.String("remote-member-id", m.ID.String()),
				zap.Error(err),
			)
			continue
		}
		var b []byte
		b, err = io.ReadAll(resp.Body)
		resp.Body.Close()
		if err != nil {
			lg.Warn(
				"failed to read body of response",
				zap.String("address", addr),
				zap.String("remote-member-id", m.ID.String()),
				zap.Error(err),
			)
			continue
		}
		var vers version.Versions
		if err = json.Unmarshal(b, &vers); err != nil {
			lg.Warn(
				"failed to unmarshal response",
				zap.String("address", addr),
				zap.String("remote-member-id", m.ID.String()),
				zap.Error(err),
			)
			continue
		}
		return &vers, nil
	}
	return nil, err
}

func promoteMemberHTTP(ctx context.Context, url string, id uint64, peerRt http.RoundTripper) ([]*membership.Member, error) {
	cc := &http.Client{Transport: peerRt}
	// TODO:重构成员http处理程序代码
	// 无法导入etcdhttp，因此手动构造url 
	requestUrl := url + "/members/promote/" + fmt.Sprintf("%d", id)
	req, err := http.NewRequest("POST", requestUrl, nil)
	if err != nil {
		return nil, err
	}
	req = req.WithContext(ctx)
	resp, err := cc.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	b, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}

	if resp.StatusCode == http.StatusRequestTimeout {
		return nil, ErrTimeout
	}
	if resp.StatusCode == http.StatusPreconditionFailed {
		// ErrMemberNotLearner和ErrLearnerNotReady具有相同的http状态代码
		if strings.Contains(string(b), ErrLearnerNotReady.Error()) {
			return nil, ErrLearnerNotReady
		}
		if strings.Contains(string(b), membership.ErrMemberNotLearner.Error()) {
			return nil, membership.ErrMemberNotLearner
		}
		return nil, fmt.Errorf("member promote: unknown error(%s)", string(b))
	}
	if resp.StatusCode == http.StatusNotFound {
		return nil, membership.ErrIDNotFound
	}

	if resp.StatusCode != http.StatusOK { // 所有其他类型的错误
		return nil, fmt.Errorf("member promote: unknown error(%s)", string(b))
	}

	var membs []*membership.Member
	if err := json.Unmarshal(b, &membs); err != nil {
		return nil, err
	}
	return membs, nil
}

// GetDegradeEnabledFromRemotePeers将获取群集的已启用降级状态。
func getDowngradeEnabledFromRemotePeers(lg *zap.Logger, cl *membership.RaftCluster, local types.ID, rt http.RoundTripper) bool {
	members := cl.Members()

	for _, m := range members {
		if m.ID == local {
			continue
		}
		enable, err := getDowngradeEnabled(lg, m, rt)
		if err != nil {
			lg.Warn("failed to get downgrade enabled status", zap.String("remote-member-id", m.ID.String()), zap.Error(err))
		} else {
			// 自“/Degrade/enabled”之后服务线性化数据，
			// 此函数在从终结点获得非错误响应后可以返回。
			return enable
		}
	}
	return false
}

// GetDegradeEnabled通过其对等项返回给定成员的降级启用状态
// 如果无法获取，则返回最后一个错误。
func getDowngradeEnabled(lg *zap.Logger, m *membership.Member, rt http.RoundTripper) (bool, error) {
	cc := &http.Client{
		Transport: rt,
	}
	var (
		err  error
		resp *http.Response
	)

	for _, u := range m.PeerURLs {
		addr := u + DowngradeEnabledPath
		resp, err = cc.Get(addr)
		if err != nil {
			lg.Warn(
				"failed to reach the peer URL",
				zap.String("address", addr),
				zap.String("remote-member-id", m.ID.String()),
				zap.Error(err),
			)
			continue
		}
		var b []byte
		b, err = io.ReadAll(resp.Body)
		resp.Body.Close()
		if err != nil {
			lg.Warn(
				"failed to read body of response",
				zap.String("address", addr),
				zap.String("remote-member-id", m.ID.String()),
				zap.Error(err),
			)
			continue
		}
		var enable bool
		if enable, err = strconv.ParseBool(string(b)); err != nil {
			lg.Warn(
				"failed to convert response",
				zap.String("address", addr),
				zap.String("remote-member-id", m.ID.String()),
				zap.Error(err),
			)
			continue
		}
		return enable, nil
	}
	return false, err
}

func convertToClusterVersion(v string) (*semver.Version, error) {
	ver, err := semver.NewVersion(v)
	if err != nil {
		// 允许输入版本format Major.Minor 
		ver, err = semver.NewVersion(v + ".0")
		if err != nil {
			return nil, ErrWrongDowngradeVersionFormat
		}
	}
	// 群集版本仅保留Major.Minor，删除补丁版本
	ver = &semver.Version{Major: ver.Major, Minor: ver.Minor}
	return ver, nil
}
